package com.cy.tempflinkcdc.mysql;

import com.cy.tempflinkcdc.datachange.DataChangeInfo;
import com.cy.tempflinkcdc.datachange.DataChangeSink;
import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.Properties;
import java.util.StringJoiner;

import static com.ververica.cdc.debezium.table.DebeziumOptions.getDebeziumProperties;

/**
 * @version 1.0
 * @description: 通过实现 ApplicationRunner 接口 将cdc变成组件引入Springboot服务中
 */
@Component
public class MysqlEventListener implements ApplicationRunner {





    private final DataChangeSink dataChangeSink;

    public MysqlEventListener(DataChangeSink dataChangeSink) {
        this.dataChangeSink = dataChangeSink;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DebeziumSourceFunction<DataChangeInfo> dataChangeInfoMySqlSource = buildDataChangeSourceLocal2();
        DataStream<DataChangeInfo> streamSource = env
                .addSource(dataChangeInfoMySqlSource, "mysql-source")
                .setParallelism(1);
        streamSource.addSink(dataChangeSink);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)));

        env.execute("mysql-stream-cdc");
    }


    @Value("${mysql.ip}")
    private String ip;
    @Value("${mysql.username}")
    private String username;
    @Value("${mysql.collection}")
    private String collection;
    @Value("${mysql.port}")
    private Integer port;
    @Value("${mysql.password}")
    private String password;


    //这边修改监听的数据库信息 MySQL
    private DebeziumSourceFunction<DataChangeInfo> buildDataChangeSourceLocal2() {
        return MySqlSource.<DataChangeInfo>builder()

                .hostname(ip)
                .port(3306)
                .username(username)
                .password(password)
                .databaseList(collection)
                // .tableList("flink-cdc-producer.producer_content", "flink-cdc-producer.name_content")
                .tableList(collection + ".*")


                //.hostname("192.168.5.17")
                //.port(3306)
                //.username("root")
                //.password("123456")
                //.databaseList("datacenter_collection")
                //// .tableList("flink-cdc-producer.producer_content", "flink-cdc-producer.name_content")
                //.tableList("datacenter_collection.*")
                /*
                 * initial初始化快照,即全量导入后增量导入(检测更新数据写入)
                 * latest:只进行增量导入(不读取历史变化)
                 * timestamp:指定时间戳进行数据导入(大于等于指定时间错读取数据)
                 */
                .startupOptions(StartupOptions.latest())
                .deserializer(new MysqlDeserialization())
                .debeziumProperties(getDebeziumProperties())
                .serverTimeZone("GMT+8")
                .build();
    }
    private static Properties getDebeziumProperties(){
        Properties properties = new Properties();
        properties.setProperty("converters", "dateConverters");
        //根据类在那个包下面修改
        properties.setProperty("dateConverters.type", "com.cy.tempflinkcdc.mysql.MySqlDateTimeConverter");
        properties.setProperty("dateConverters.format.date", "yyyy-MM-dd");
        properties.setProperty("dateConverters.format.time", "HH:mm:ss");
        properties.setProperty("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss");
        properties.setProperty("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss");
        properties.setProperty("dateConverters.format.timestamp.zone", "UTC+8");
        properties.setProperty("debezium.snapshot.locking.mode","none"); //全局读写锁，可能会影响在线业务，跳过锁设置
        properties.setProperty("include.schema.changes", "true");
        properties.setProperty("bigint.unsigned.handling.mode","long");
        properties.setProperty("decimal.handling.mode","double");
        return properties;
    }



    private DebeziumSourceFunction<DataChangeInfo> buildDataChangeSourceLocal() {
        return MySqlSource.<DataChangeInfo>builder()
                .hostname("192.168.1.12")
                .port(3306)
                .username("root")
                .password("root")
                .databaseList("datacenter_collection")
                // .tableList("flink-cdc-producer.producer_content", "flink-cdc-producer.name_content")
                .tableList("datacenter_collection.*")
                /*
                 * initial初始化快照,即全量导入后增量导入(检测更新数据写入)
                 * latest:只进行增量导入(不读取历史变化)
                 * timestamp:指定时间戳进行数据导入(大于等于指定时间错读取数据)
                 */
                .startupOptions(StartupOptions.latest())
                .deserializer(new MysqlDeserialization())
                .serverTimeZone("GMT+8")
                .build();
    }

    private DebeziumSourceFunction<DataChangeInfo> buildDataChangeSourceRemote() {
        StringJoiner tableJoiner = new StringJoiner(",");
        for (int i = 1; i <= 31; i++) {
            tableJoiner.add("migration_tables.table_" + i);
        }
        return MySqlSource.<DataChangeInfo>builder()
                .hostname("117.72.32.234")
                .port(3306)
                .username("root")
                .password("Zz24270507..")
                .databaseList("migration_tables")
                .tableList(tableJoiner.toString())
                /*
                 * initial初始化快照,即全量导入后增量导入(检测更新数据写入)
                 * latest:只进行增量导入(不读取历史变化)
                 * timestamp:指定时间戳进行数据导入(大于等于指定时间错读取数据)
                 */
                .startupOptions(StartupOptions.latest())
                .deserializer(new MysqlDeserialization())
                .serverTimeZone("GMT+8")
                .build();
    }
}
